package com.ttx.zookeeper.lock.raw;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * @author TimFruit
 * @date 20-2-28 下午1:54
 */
public abstract class AbstractLock implements Lock{
    protected ZooKeeper zk;
    protected String lockNodePath;


    protected String guid;

    protected String spilt="_";

    protected String mark;

    protected String lockNodePrefix;
    protected String lockNodeName;


    public AbstractLock(ZooKeeper zk, String lockNodePath, String mark) {
        if(mark==null || mark.trim().length()==0){
            throw new RuntimeException("mark不能为空");
        }
        this.zk=zk;
        this.lockNodePath=lockNodePath;
        this.guid=UUID.randomUUID().toString();
        this.mark=mark;
        this.lockNodePrefix=guid+spilt+mark+spilt;
    }


    @Override
    public void lock() {
        try {
            zk.create(lockNodePath+"/"+lockNodePrefix,
                    guid.getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);

            while(true){

                List<String> children= null;
                try {
                    children = zk.getChildren(lockNodePath, false);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }

                //根据序号排序
                sortedBySequence(children);

                if(isLocked(children)){
                    return;
                }

                //监控第一个节点,注册watcher,watcher被触发一次后就会失效
                final CountDownLatch cdl=new CountDownLatch(1);
                String firstChild=children.get(0);
                String firstChildPath=lockNodePath+"/"+firstChild;
                Stat stat=zk.exists(firstChildPath, (watchedEvent -> cdl.countDown()));

                if(stat!=null){
                    cdl.await();
                }
            }


        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    @Override
    public void unLock() {
        if(lockNodeName==null || lockNodeName.trim().length()==0){
            throw new RuntimeException("未获取锁(lock()), 不能调用本方法");
        }
        try {
            zk.delete(lockNodePath+"/"+lockNodeName, 0);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }



    /**
     * 判断是否获取到锁
     *
     * 成功索取锁之后，需要设置{@link #lockNodeName}, 用于释放锁{@link #unLock()}
     * @param children 已根据序号从小到大排序的节点
     * @return
     */
    protected boolean isLocked(List<String> children){
        String firstChild=children.get(0);
        if(firstChild.startsWith(lockNodePrefix)){
            lockNodeName=firstChild;
            System.out.println("获取锁"+lockNodeName);
            return true;
        }
        return false;
    }



    protected void sortedBySequence(List<String> children){
        // child, seq
        Map<String, String> childSequenceMap=children.stream().collect(Collectors.toMap(
                s-> s,
                s -> {
                    // guid+spilt+mark+spilt+Sequence;
                    String[] strs=s.split(spilt);
                    String seq=strs[2];
                    return seq;
                }
        ));

        Collections.sort(children, (s1, s2)->{
            String seq1=childSequenceMap.get(s1);
            String seq2=childSequenceMap.get(s2);
            return seq1.compareTo(seq2);
        });

    }

}